package net.maergaiyun.func;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import net.maergaiyun.util.TimeUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;


@Slf4j
public class VisitorMapFunction extends RichMapFunction<JSONObject, String> {

    /**
     * 记录用户的udid访问
     */
    private ValueState<String> newDayVisitorState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 状态初始化
        newDayVisitorState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newDayVisitorState", String.class));
    }

    @Override
    public String map(JSONObject value) throws Exception {
        // 之前的访问日期
        String beforeDateState = newDayVisitorState.value();

        // 当前的访问时间戳
        Long ts = value.getLong("ts");
        String currentDateStr = TimeUtil.format(ts);

        if (StringUtils.isNotBlank(beforeDateState) && beforeDateState.equalsIgnoreCase(currentDateStr)) {
            value.put("is_new", 0);
            log.info("老访客, {}", currentDateStr);
        } else {
            value.put("is_new", 1);
            newDayVisitorState.update(currentDateStr);
            log.info("新访客, {}", currentDateStr);
        }

        return value.toJSONString();
    }
}
